package com.staryea.kafkatomysql.handler;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.staryea.kafkatomysql.mapper.KafkaToMysqlMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author gonghl@staryea.com
 * @since 2022/6/12
 */
@Slf4j
@Component
public class KafkaToMysqlHandler {

    @Autowired
    private KafkaToMysqlMapper mapper;

    @Value("${kafkaToMysql.table}")
    private String table;

    private LinkedBlockingQueue<Map<String, Object>> queue = new LinkedBlockingQueue<>(2000);

    private ExecutorService pool;

    private volatile boolean isTerminal;

    @PostConstruct
    public void init() {
        pool = Executors.newFixedThreadPool(10);
        new Thread(this::startConsumer).start();

    }

    @KafkaListener(topics = "${kafkaToMysql.topic}")
    public void processMessage(String content) {
        try {
            JSONObject jsonObject = JSONObject.parseObject(content);
            Map<String, Object> data = convertData(jsonObject);
            if (data == null) {
                return;
            }

            queue.put(data);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private Map<String, Object> convertData(JSONObject jsonObject) {
        JSONArray data = jsonObject.getJSONArray("data");
        if (CollectionUtils.isEmpty(data)) {
            return null;
        }

        Map<String, Object> map = new HashMap<>(data.size());
        for (int i = 0; i < data.size(); i++) {
            JSONObject object = data.getJSONObject(i);
            map.putAll(object);
        }

        return map;
    }

    private void startConsumer() {
        long start = System.currentTimeMillis();
        long end = System.currentTimeMillis();

        while (isTerminal) {
            long interval = end - start;
            if (queue.size() > 1000 || (queue.size() > 0 && interval > 1000)) {
                List<Map<String, Object>> temp = new ArrayList<>();
                queue.drainTo(temp);
                saveToDb(temp);
                start = System.currentTimeMillis();
                end = start;
                continue;
            }

            try {
                Thread.sleep(200L);
            } catch (Exception ignored) {}
            end = System.currentTimeMillis();
        }
    }

    private synchronized void saveToDb(List<Map<String, Object>> list) {
        pool.execute(() -> mapper.batchAddData(list, table));
    }

    @PreDestroy
    public void onDestroy() {
        isTerminal = true;
    }

}
